package com.atguigu.sink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

import java.util.Objects;

/**
 * @author fuc
 * @create 2020-06-03
 */
@Slf4j
public class MySink extends AbstractSink implements Configurable {

  /** 定义全局的前缀&后缀 */
  private String prefix;

  private String subfix;

  @Override
  public void configure(Context context) {
    // 读取配置信息给前后缀赋值
    context.getString("prefix");
    context.getString("subfix", "atguigu");
  }

  /** 1.获取Channel 2.从Channel获取事务以及数据 3.发送数据 */
  @Override
  public Status process() throws EventDeliveryException {
    // 1.定义返回值
    Status status = null;

    // 2.获取Channel
    Channel channel = getChannel();

    // 3.从Channel获取事务
    Transaction transaction = channel.getTransaction();

    // 4.开启事务
    transaction.begin();

    try {
      // 5.从Channel获取数据
      Event event = channel.take();

      // 6.处理事件
      if (Objects.nonNull(event)) {
        String body = new String(event.getBody());
        log.info(prefix + "--" + body + "--" + subfix);
        // 7.提交事务
        transaction.commit();
        transaction.close();
      }
      // 成功提交，修改状态信息
      status = Status.READY;
    } catch (ChannelException e) {
      e.printStackTrace();
      // 9.提交事务失败
      transaction.rollback();

      // 10.修改状态
      status = Status.BACKOFF;
    } finally {
      // 11.最终，关闭事务
      transaction.close();
    }
    // 12.返回状态信息
    return status;
  }
}
